package io.smallrye.mutiny.groups;

import static io.smallrye.mutiny.helpers.ParameterValidation.nonNull;

import java.time.Duration;
import java.util.Optional;
import java.util.concurrent.CompletionException;

import io.smallrye.common.annotation.CheckReturnValue;
import io.smallrye.mutiny.Context;
import io.smallrye.mutiny.TimeoutException;
import io.smallrye.mutiny.Uni;
import io.smallrye.mutiny.infrastructure.Infrastructure;
import io.smallrye.mutiny.operators.uni.UniBlockingAwait;
import io.smallrye.mutiny.operators.uni.builders.UniCreateFromKnownFailure;
import io.smallrye.mutiny.operators.uni.builders.UniCreateFromKnownItem;

/**
 * Waits and returns the item emitted by the {@link Uni}. If the {@link Uni} receives a failure, the failure is thrown.
 * <p>
 * This class lets you configure how to retrieves the item of a {@link Uni} by blocking the caller thread.
 *
 * @param <T> the type of item
 * @see Uni#await()
 */
public class UniAwait<T> {

    private final Uni<T> upstream;
    private final Context context;

    public UniAwait(Uni<T> upstream, Context context) {
        this.upstream = nonNull(upstream, "upstream");
        this.context = context;
    }

    /**
     * Subscribes to the {@link Uni} and waits (blocking the caller thread) <strong>indefinitely</strong> until a
     * {@code item} event is fired or a {@code failure} event is fired by the upstream uni.
     * <p>
     * If the {@link Uni} fires an item, it returns that item, potentially {@code null} if the operation
     * returns {@code null}.
     * If the {@link Uni} fires a failure, the original exception is thrown (wrapped in
     * a {@link java.util.concurrent.CompletionException} it's a checked exception).
     * <p>
     * Note that each call to this method triggers a new subscription.
     *
     * @return the item from the {@link Uni}, potentially {@code null}
     */
    public T indefinitely() {
        return atMost(null);
    }

    /**
     * Subscribes to the {@link Uni} and waits (blocking the caller thread) <strong>at most</strong> the given duration
     * until an item or failure is fired by the upstream uni.
     * <p>
     * If the {@link Uni} fires an item, it returns that item, potentially {@code null} if the operation
     * returns {@code null}.
     * If the {@link Uni} fires a failure, the original exception is thrown (wrapped in
     * a {@link java.util.concurrent.CompletionException} it's a checked exception).
     * If the timeout is reached before completion, a {@link TimeoutException} is thrown.
     * <p>
     * Note that each call to this method triggers a new subscription.
     *
     * @param duration the duration, must not be {@code null}, must not be negative or zero.
     * @return the item from the {@link Uni}, potentially {@code null}
     */
    public T atMost(Duration duration) {
        // Optimize the case where the item/failure is known
        // and subscription/synchronization is not needed
        if (upstream instanceof UniCreateFromKnownItem<T> known) {
            return awaitKnownItem(known, duration);
        } else if (upstream instanceof UniCreateFromKnownFailure<T> known) {
            awaitKnownFailure(known, duration);
        }
        return UniBlockingAwait.await(upstream, duration, context);
    }

    /**
     * Indicates that you are awaiting for the item event of the attached {@link Uni} wrapped into an {@link Optional}.
     * So if the {@link Uni} fires {@code null} as item, you receive an empty {@link Optional}.
     *
     * @return the {@link UniAwaitOptional} configured to produce an {@link Optional}.
     */
    @CheckReturnValue
    public UniAwaitOptional<T> asOptional() {
        return new UniAwaitOptional<>(upstream, context);
    }

    private T awaitKnownItem(UniCreateFromKnownItem<T> known, Duration duration) {
        validateDuration(duration);
        // Blocking should not matter in this case but we retain the check for backward compatibility
        if (!Infrastructure.canCallerThreadBeBlocked()) {
            throw UniBlockingAwait.currentThreadCannotBeBlocked();
        }
        return known.getItem();
    }

    private void awaitKnownFailure(UniCreateFromKnownFailure<T> known, Duration duration) {
        validateDuration(duration);
        // Blocking should not matter in this case but we retain the check for backward compatibility
        if (!Infrastructure.canCallerThreadBeBlocked()) {
            throw UniBlockingAwait.currentThreadCannotBeBlocked();
        }
        Throwable throwable = known.getFailure();
        if (throwable instanceof RuntimeException) {
            throw (RuntimeException) throwable;
        }
        throw new CompletionException(throwable);
    }

    private void validateDuration(Duration duration) {
        if (duration != null && (duration.isZero() || duration.isNegative())) {
            throw new IllegalArgumentException("`duration` must be greater than zero");
        }
    }

}
